package cas.ihep.hadoop.tr;

import cas.ihep.util.Tuple;
import com.google.common.base.Strings;
import com.google.common.io.Closer;
import org.apache.commons.cli.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.*;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class Check implements Closeable{

    private ExecutorService pool;
    private ThreadLocal<byte[]> tlsStorage;

    private Check(int thrs){
        pool= Executors.newFixedThreadPool(thrs);
        tlsStorage=new ThreadLocal<>();
    }

    private void check(File localFile, Path hFile, FileSystem hdfs)throws IOException{
        boolean f1=localFile.isDirectory();
        boolean f2=hdfs.isDirectory(hFile);
        if(f1 && f2){
            HashMap<String,Tuple<File,Path>> fileMap=new HashMap<>();
            File[] subfiles=localFile.listFiles();
            FileStatus []subhfiles=hdfs.listStatus(hFile);
            if(subfiles!=null && subhfiles!=null){
                for(File file:subfiles){
                    fileMap.put(file.getName(),new Tuple<>(file,null));
                }
                for(FileStatus file:subhfiles){
                    Path path=file.getPath();
                    String filename=path.getName();
                    Tuple<File,Path> entry=fileMap.get(filename);
                    if(entry!=null && entry.t2==null){
                        entry.t2=path;
                    }else if(entry==null){
                        System.err.println("File mismatch: "+path.toString());
                    }
                }
                for(Map.Entry<String,Tuple<File,Path>> entry:fileMap.entrySet()){
                    Tuple<File,Path> tuple=entry.getValue();
                    if(tuple.t2==null){
                        System.err.println("File mismatch: "+tuple.t1.getCanonicalPath());
                        continue;
                    }
                    check(tuple.t1,tuple.t2,hdfs);
                }
            }
        }else if(!f1 && !f2){
            if(localFile.length()!=hdfs.getFileStatus(hFile).getLen()){
                System.err.println("File mismatch: "+localFile.getCanonicalPath()+'&'+hFile.toString());
                return;
            }
            pool.submit(new Callable<Integer>() {
                @Override
                public Integer call() throws Exception {
                    byte[] buf=tlsStorage.get();
                    if(buf==null){
                        buf=new byte[8192];
                        tlsStorage.set(buf);
                    }
                    try(InputStream lis=new FileInputStream(localFile);InputStream his=hdfs.open(hFile)){
                        int each1,each2;
                        while(true){
                            each1=lis.read(buf,0,4096);
                            each2=his.read(buf,4096,4096);
                            if(each1!=each2){
                                System.err.println("File mismatch: "+localFile.getCanonicalPath()+'&'+hFile.toString());
                                break;
                            }else if(each1==-1){
                                break;
                            }else{
                                for(int i=0;i<each1;i++){
                                    if(buf[i]!=buf[4096+i]){
                                        System.err.println("File mismatch: "+localFile.getCanonicalPath()+'&'+hFile.toString());
                                        break;
                                    }
                                }
                            }
                        }
                    }
                    return 0;
                }
            });
        }else{
            System.err.println("File type mismatch: "+localFile.getCanonicalPath()+'&'+hFile.toString());
        }
    }

    private static int optThreads(CommandLine  cmdline){
        if(cmdline.hasOption('t')){
            return Integer.parseInt(cmdline.getOptionValue('t'));
        }
        return 1;
    }

    private static String optHadoopHome(CommandLine cmdline){
        if(cmdline.hasOption('c')){
            return cmdline.getOptionValue('c');
        }
        return Strings.nullToEmpty(System.getenv("HADOOP_HOME"));
    }

    public static void main(String args[]) throws IOException, ParseException {
        Options opts=new Options();
        opts.addOption("t","thread",true,"Specify the number of threads");
        opts.addOption("h","help",false,"Show this message");
        opts.addOption("c","conf",true,"Specify the config directory of hadoop");
        try(Closer closer=Closer.create()){
            CommandLine cmdline=new DefaultParser().parse(opts,args);
            List<String> argv=cmdline.getArgList();
            int sz=argv.size();
            if(sz!=2 || cmdline.hasOption('h')){
                new HelpFormatter().printHelp("hadoop-check","",opts,null,true);
                return;
            }
            Configuration hConf=new Configuration();
            String hdpHome=optHadoopHome(cmdline);
            hConf.addResource(closer.register(new FileInputStream(hdpHome+"/etc/hadoop/core-site.xml")));
            hConf.addResource(closer.register(new FileInputStream(hdpHome+"/etc/hadoop/hdfs-site.xml")));
            FileSystem hdfs=closer.register(FileSystem.get(hConf));
            Check check=closer.register(new Check(optThreads(cmdline)));
            File f1=new File(argv.get(0)).getCanonicalFile();
            Path p1=new Path(argv.get(1));
            if(!f1.getName().equals(p1.getName())){
                System.err.println("Warning: name mismatch");
            }
            check.check(f1,p1,hdfs);
        }
    }

    @Override
    public void close() {
        pool.shutdown();
        try {
            pool.awaitTermination(100000, TimeUnit.DAYS);
        } catch (InterruptedException ignored) {
        }
    }
}
